package com.example.toy;

import com.example.toy.core.FsOfflineToySystemFactoryFactory;
import com.example.toy.core.ToySystemFactoryProducer;
import com.example.toy.core.commit.CommitMsg;
import com.example.toy.core.commit.CommitState;
import com.example.toy.core.commit.CommitType;
import com.example.toy.core.commit.ToyCommitAbstractFactory;
import com.example.toy.core.commit.ToyCommitView;
import com.example.toy.core.commit.ToyCommitter;
import com.example.toy.core.conf.ToyConfigs;
import com.example.toy.core.conf.ToySystemConfig;
import com.example.toy.core.conf.ToyTableConfig;
import com.example.toy.core.config.FsOfflineSystemConfigs;
import com.example.toy.core.exception.NotExistsException;
import com.example.toy.core.exception.TableCreateException;
import com.example.toy.core.locker.ToyLock;
import com.example.toy.core.system.AbstractToySystem;
import com.example.toy.core.table.FsOfflineTableMeta;
import com.example.toy.core.table.FsOfflineToyTable;
import com.example.toy.core.table.TableIdentifier;
import com.example.toy.core.table.ToyTable;
import com.example.toy.core.utils.TableMetaHelper;
import com.example.toy.core.utils.TablePatternUtils;
import com.example.toy.core.utils.ToyPathDefine;
import com.example.toy.core.version.ToyVersion;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.regex.Pattern;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * 基于文件路径管理一个或多个toy表。
 *
 * <p>在FsOfflineToySystemImpl中，表的catalog和database默认是空的，不起作用
 *
 * <p>检查根目录下：
 * <li>如果存在元信息文件夹，抛出异常
 * <li>如果不存在元信息文件夹，根目录下的所有文件夹，尝试作为toy表加载，每个子文件夹如果存在toy元信息文件夹，则作为toy表，否则忽略该文件夹
 */
@Slf4j
public class FsOfflineToySystemImpl extends AbstractToySystem implements ToyExternalSystem {

  private final Path path;
  private final FileSystem fs;
  private final FsOfflineToySystemFactoryFactory factory;

  @SneakyThrows
  public FsOfflineToySystemImpl(ToySystemConfig config) {
    super(config);
    this.path = new Path(config.getString(FsOfflineSystemConfigs.PATH));
    this.fs = this.path.getFileSystem(new Configuration());
    validatePath();

    this.factory = new FsOfflineToySystemFactoryFactory(this);
    log.info("FsOfflineToySystem open at path: " + this.path);
  }

  private void validatePath() throws IllegalArgumentException, IOException {
    if (fs.exists(path)) {
      if (fs.isFile(path)) {
        throw new IllegalArgumentException(
            "Unsupported FsOfflineToySystem path at file already exists: " + path);
      }
    } else {
      fs.mkdirs(path);
    }
    if (TableMetaHelper.isTablePath(path)) {
      throw new IllegalArgumentException(
          "FsOfflineToySystem can not create at a toy table path: " + path);
    }
  }

  @SneakyThrows
  @Override
  public List<TableIdentifier> list(String catalog, String database, String tablePattern) {
    Pattern pattern = TablePatternUtils.mkPattern(tablePattern);

    FileStatus[] statuses = fs.listStatus(this.path);
    List<TableIdentifier> toyTables = new ArrayList<>();
    for (FileStatus status : statuses) {
      Path basePath = status.getPath();
      if (TableMetaHelper.isTablePath(basePath)) {
        Path toyConfigPath = ToyPathDefine.getTableConfigFilePath(basePath);
        ToyTableConfig config = ToyTableConfig.fromCacheOrPath(toyConfigPath);
        String tableName = config.getTableName();

        if (TablePatternUtils.match(pattern, tableName)) {
          toyTables.add(new TableIdentifier(tableName));
        }
      }
    }
    return toyTables;
  }

  @Override
  @SneakyThrows
  public Path getToyTablePath(TableIdentifier tableIdentifier) throws NotExistsException {
    FileStatus[] statuses = fs.listStatus(this.path);
    for (FileStatus status : statuses) {
      Path basePath = status.getPath();
      if (TableMetaHelper.isTablePath(basePath)) {
        Path toyConfigPath = ToyPathDefine.getTableConfigFilePath(basePath);
        ToyTableConfig config = ToyTableConfig.fromCacheOrPath(toyConfigPath);
        if (tableIdentifier.getTable().equals(config.getTableName())) {
          return basePath;
        }
      }
    }

    throw new NotExistsException("Toy table not exists: " + tableIdentifier);
  }

  @Override
  public ToyTable loadTable(Path dir, ToyVersion version) {
    return new FsOfflineToyTable(this, dir, version);
  }

  @Override
  public ToyTable getToyTable(TableIdentifier tableIdentifier, ToyVersion version) {
    Path toyTablePath = getToyTablePath(tableIdentifier);
    return loadTable(toyTablePath, version);
  }

  @Override
  @SneakyThrows
  public ToyTable createTable(Path basePath, ToyTableConfig config) {
    if (fs.exists(basePath)) {
      if (TableMetaHelper.isTablePath(basePath)) {
        Path tableConfigFilePath = ToyPathDefine.getTableConfigFilePath(basePath);
        ToyTableConfig existedConfig = ToyTableConfig.fromCacheOrPath(tableConfigFilePath);
        if (TableMetaHelper.isSameTable(existedConfig, config)) {
          return loadTable(basePath, ToyVersion.LATEST_VERSION);
        }
      } else if (isEmptyDir(fs, basePath)) {
        // continue to create table
      } else {
        throw new IllegalArgumentException("table path already been used: " + basePath);
      }
    }

    TableIdentifier tableIdentifier = new TableIdentifier(config.getTableName());
    ToyLock tableLock = factory.buildLockFactory().buildTableLock(tableIdentifier, basePath);

    try {
      try {
        tableLock.lockQuickly();

        ToyCommitAbstractFactory commitFactory = factory.buildCommitFactory();
        ToyCommitter committer = commitFactory.getCommitter(tableIdentifier, basePath);

        CommitMsg initCommit =
            CommitMsg.builder()
                .msg("init table")
                .commitType(CommitType.INIT)
                .commitState(CommitState.REQUESTED)
                .schema(config.getSchema())
                .version(ToyVersion.FIRST_VERSION)
                .build();

        committer.submit(initCommit, CommitState.INFLIGHT);

        Path metaDirPath = ToyPathDefine.getMetaDirPath(basePath);
        fs.mkdirs(metaDirPath);

        try (FSDataOutputStream configStream =
            fs.create(ToyPathDefine.getTableConfigFilePath(basePath), true); ) {
          config.store(configStream);
        }

        committer.changeState(initCommit.getVersion(), CommitState.SUCCEED);
      } catch (Exception e) {
        deleteToyMetaDir(tableIdentifier, basePath);
        throw new TableCreateException("Failed to create table: " + tableIdentifier, e);
      } finally {
        tableLock.unlock();
      }
    } catch (Exception e) {
      try {
        // 删除表文件夹
        fs.delete(basePath, true);
      } catch (Exception exception) {
        log.warn("Can not delete create failed table path: " + basePath);
      }
      throw e;
    }

    return new FsOfflineToyTable(this, basePath, ToyVersion.FIRST_VERSION);
  }

  @SneakyThrows
  private boolean isEmptyDir(FileSystem fs, Path basePath) {
    if (fs.exists(basePath)) {
      FileStatus fileStatus = fs.getFileStatus(basePath);
      return fileStatus.isDirectory() && fs.listStatus(basePath).length == 0;
    } else {
      return false;
    }
  }

  /**
   * 删除 .toy 文件夹
   *
   * @param basePath
   */
  @SneakyThrows
  private void deleteToyMetaDir(TableIdentifier tableIdentifier, Path basePath) {
    //    删除 .toy 元信息文件夹，可以省略清理commits的步骤
    //    ToyCommitAbstractFactory commitFactory = factory.buildCommitFactory();
    //    commitFactory.dropTableCommits(tableIdentifier, basePath);

    Path metaDirPath = ToyPathDefine.getMetaDirPath(basePath);
    fs.delete(metaDirPath, true);
  }

  @SneakyThrows
  @Override
  public ToyTable create(TableIdentifier tableIdentifier, ToyTableConfig config) {
    config.setIfMissing(ToyConfigs.TABLE_NAME.name(), tableIdentifier.getTable());
    if (!Objects.equals(config.getTableName(), tableIdentifier.getTable())) {
      throw new IllegalArgumentException(
          "Table identifier name must be equal to the table name in table config");
    }

    Path tableBasePath = new Path(this.path, config.getTableName());
    return createTable(tableBasePath, config);
  }

  @SneakyThrows
  @Override
  public void drop(TableIdentifier tableIdentifier) {
    ToyTable toyTable = getToyTable(tableIdentifier);
    toyTable.drop();
    fs.delete(toyTable.getTableMeta().getBasePath(), true);
  }

  @Override
  public ToySystemFactoryProducer getSystemFactoryProducer() {
    return factory;
  }

  @Override
  public void archive(TableIdentifier tableIdentifier, ToyVersion version) {}

  @Override
  public List<ToyVersion> getAvailableArchiveVersions(TableIdentifier tableIdentifier) {
    return null;
  }

  @Override
  public void removeArchive(TableIdentifier tableIdentifier, ToyVersion version) {}
}
